package com.shengzai.flink.core

import org.apache.flink.streaming.api.functions.{KeyedProcessFunction, ProcessFunction}
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import scala.collection.mutable

object Demo7ProcessFunction {
  def main(args: Array[String]): Unit = {

    /**
     * process: 是flink底层的一个算子，传入一条可以返回多条，再process里面可以操作flink的数据，时间和状态
     * 用法
     * 1、再普通的DataStream后面使用，需要一个ProcessFunction
     */
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(2)

    val socketDS: DataStream[String] = env.socketTextStream("master", 8888)

    val processDS: DataStream[String] = socketDS.process(new ProcessFunction[String, String] {
      /**
       *
       * @param value :一行数据
       * @param ctx   : 上下文对象
       * @param out   :发送到下游
       */
      override def processElement(value: String,
                                  ctx: ProcessFunction[String, String]#Context,
                                  out: Collector[String]): Unit = {

        val split: Array[String] = value.split(",")
        for (elem <- split) {
          out.collect(elem)
        }
      }
    })

    val keyByDS: KeyedStream[String, String] = processDS.keyBy(word => word)

    //定义一个map，一个单词对应一个key，单词数量为value
    val hashMap = new mutable.HashMap[String, Int]()
    /**
     * 2、再keyBy之后使用：需要一个KeyedProcessFunction
     */
    val tmpDS: DataStream[(String, Int)] = keyByDS.process(
      new KeyedProcessFunction[String, String, (String, Int)] {
        override def processElement(value: String,
                                    ctx: KeyedProcessFunction[String, String, (String, Int)]#Context,
                                    out: Collector[(String, Int)]): Unit = {
          var count: Int = hashMap.getOrElse(value, 0)

          count = count + 1

          hashMap.put(value, count)

          out.collect(value, count)
        }
      }
    )
    // tmpDS.print()

    /**
     * 3、再window之后使用,需要一个ProcessWindowFunction
     */

    val windowsDS: WindowedStream[String, String, TimeWindow] = keyByDS.window(
      TumblingProcessingTimeWindows.of(Time.seconds(10))
    )


    val windowsRes: DataStream[(String, Int)] = windowsDS.process(new ProcessWindowFunction[String, (String, Int), String, TimeWindow] {
      /**
       * process: 每一个窗口执行一次，会增和窗口的数据都传到process方法中
       *
       * @param key      ：分组的key
       * @param context  :上下文对象, 获取到窗口的开始和结束时间
       * @param elements ： 窗口内所有的数据
       * @param out      ：用于将数据发送到下游
       */
      override def process(key: String,
                           context: Context,
                           elements: Iterable[String],
                           out: Collector[(String, Int)]): Unit = {
        val count: Int = elements.size

        out.collect(key, count)
      }
    })
    windowsRes.print()




    env.execute()


  }

}
